package org.aurora.core.server.scheduler;

import org.aurora.common.thread.DaemonThread;
import org.aurora.model.entity.ScheduleTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

/**
 * clear()
 * 以原子方式从此队列中删除所有元素
 * drainTo(Collection<? super E> c)
 * 从此队列中删除所有可用元素，并将它们添加到给定的集合中。
 * offer(E e)
 * 如果可以立即插入指定的元素，而不会超出队列的容量，则在此队列的尾部插入指定的元素，成功时返回 true ，并且 false 此队列已满。
 * offer(E e, long timeout, TimeUnit unit)
 * 在此队列的尾部插入指定的元素，如有必要，等待指定的等待时间，以便空间变为可用。
 * peek()
 * 检索但不删除此队列的头部，如果此队列为空，则返回 null 。
 * poll()
 * 检索并删除此队列的头部，如果此队列为空，则返回 null 。
 * poll(long timeout, TimeUnit unit)
 * 检索并删除此队列的头部，并在必要时等待指定的等待时间，以便元素可用。
 * remainingCapacity()
 * 返回此队列理想情况下（在没有内存或资源约束的情况下）可以接受而不阻塞的附加元素的数量。
 * take()
 * 检索并删除此队列的头部，必要时等待，直到元素可用。
 */
public class FastSchedulerTaskBlockingQueue {
    private static final Logger log = LoggerFactory.getLogger(FastSchedulerTaskBlockingQueue.class);

    private BlockingDeque<ScheduleTask> fastScheduleTaskQueue;
    private Thread executeThread;
    private Consumer<ScheduleTask> schedulerTaskConsumer;

    private final AtomicBoolean isStart = new AtomicBoolean(false);
    private volatile boolean isStop = false;

    public FastSchedulerTaskBlockingQueue(Consumer<ScheduleTask> schedulerTaskConsumer) {
        this.fastScheduleTaskQueue = new LinkedBlockingDeque<>();
    }

    public FastSchedulerTaskBlockingQueue(Consumer<ScheduleTask> schedulerTaskConsumer, int queueSize) {
        this.fastScheduleTaskQueue = new LinkedBlockingDeque<>(queueSize);
        this.executeThread = new DaemonThread();
    }

    public void offerTask(ScheduleTask schedulerTask) {
        this.fastScheduleTaskQueue.offer(schedulerTask);
    }

    public boolean offerTask(ScheduleTask schedulerTask, long timeout, TimeUnit unit) throws InterruptedException {
        return this.fastScheduleTaskQueue.offer(schedulerTask, timeout, unit);
    }

    public void clearAllTask() {
        this.fastScheduleTaskQueue.clear();
    }

    public ScheduleTask takeTask() throws InterruptedException {
        return this.fastScheduleTaskQueue.take();
    }

    public ScheduleTask pollTask() {
        return this.fastScheduleTaskQueue.poll();
    }

    public ScheduleTask pollTask(long timeout, TimeUnit unit) throws InterruptedException {
        return this.fastScheduleTaskQueue.poll(timeout, unit);
    }

    public void start() {
        if (isStart.compareAndSet(false, true)) {
            this.executeThread = new DaemonThread(() -> {
                while (true) {
                    try {
                        ScheduleTask schedulerTask = this.fastScheduleTaskQueue.take();

                        if (schedulerTask == null) {
                            continue;
                        }
                        // todo 是否需要加限流
                        this.schedulerTaskConsumer.accept(schedulerTask);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, "fastSchedulerTaskBlockingQueueThread");
            this.executeThread.start();
            log.info("FastSchedulerTaskBlockingQueue started");
        } else {
            log.warn("FastSchedulerTaskBlockingQueue already started");
        }
    }

    public void stop() {
        this.isStop = true;
        this.fastScheduleTaskQueue.clear();
    }
}
